use futures_util::TryStreamExt;
use mongodb::{
    bson::{Bson, Document},
    Client, Collection, Database,
};
use std::sync::{Arc, Mutex, OnceLock};
use tracing::{error, info};
use anyhow::{Result, anyhow};

use crate::config::Config;
use crate::storage::storage_provider::{IStorage, IStorageExt, ICollection};

/// MongoDB存储管理器
#[derive(Debug, Clone)]
pub struct MongoStorage {
    client: Arc<MongoClientWrapper>,
}

/// MongoDB集合实现
#[derive(Debug)]
struct MongoCollectionWrapper<T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync> {
    collection: Collection<T>,
}

#[async_trait::async_trait]
impl<T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync> ICollection<T> for MongoCollectionWrapper<T> {
    async fn insert_one(&self, document: T) -> Result<()> {
        self.collection.insert_one(document, None).await?;
        Ok(())
    }
    
    async fn insert_many(&self, documents: Vec<T>) -> Result<()> {
        self.collection.insert_many(documents, None).await?;
        Ok(())
    }
    
    async fn find(&self, filter: Document) -> Result<Vec<T>> {
        let cursor = self.collection.find(filter, None).await?;
        let results: Vec<T> = cursor.try_collect().await?;
        Ok(results)
    }
    
    async fn find_one(&self, filter: Document) -> Result<Option<T>> {
        let result = self.collection.find_one(filter, None).await?;
        Ok(result)
    }
    
    async fn update_one(&self, filter: Document, update: Document) -> Result<()> {
        self.collection.update_one(filter, update, None).await?;
        Ok(())
    }
    
    async fn update_many(&self, filter: Document, update: Document) -> Result<()> {
        self.collection.update_many(filter, update, None).await?;
        Ok(())
    }
    
    async fn delete_one(&self, filter: Document) -> Result<()> {
        self.collection.delete_one(filter, None).await?;
        Ok(())
    }
    
    async fn aggregate(&self, pipeline: Vec<Document>) -> Result<Vec<Document>> {
        let cursor = self.collection.aggregate(pipeline, None).await?;
        let results: Vec<Document> = cursor.try_collect().await?;
        Ok(results)
    }
}

#[derive(Debug)]
struct MongoClientWrapper {
    client: Client,
}

impl MongoClientWrapper {
    async fn new(connection_string: &str) -> Result<Self> {
        let client = Client::with_uri_str(connection_string).await?;
        Ok(Self { client })
    }
}

// 全局单例
static MONGO_STORAGE: OnceLock<Arc<Mutex<Option<MongoStorage>>>> = OnceLock::new();

impl MongoStorage {
    /// 创建新的MongoDB存储实例
    pub async fn new(connection_string: &str) -> Result<Self> {
        let client_wrapper = MongoClientWrapper::new(connection_string).await?;
        Ok(Self {
            client: Arc::new(client_wrapper),
        })
    }
    
    /// 获取MongoStorage单例实例
    pub async fn instance() -> Result<Arc<MongoStorage>> {
        let storage = MONGO_STORAGE.get_or_init(|| {
            Arc::new(Mutex::new(None))
        });
        
        let mut guard = storage.lock().unwrap();
        
        if guard.is_none() {
            let config = Config::load()?;
            let connection_string = config.db_connection_string();
            info!("连接MongoDB: {}", connection_string);
            
            let client_wrapper = MongoClientWrapper::new(&connection_string).await?;
            *guard = Some(MongoStorage {
                client: Arc::new(client_wrapper),
            });
        }
        
        Ok(Arc::new(guard.as_ref().unwrap().clone()))
    }
    
    /// 获取数据库实例
    fn get_database(&self, db_name: &str) -> Database {
        self.client.client.database(db_name)
    }
    
    /// 获取集合（内部实现）
    fn inner_get_collection<T>(&self, db_name: &str, collection_name: &str) -> Collection<T> {
        self.get_database(db_name).collection(collection_name)
    }
    
    /// 获取文档集合（内部实现）
    fn inner_get_document_collection(&self, db_name: &str, collection_name: &str) -> Collection<Document> {
        self.inner_get_collection(db_name, collection_name)
    }
    
    /// 运行MongoDB命令（内部实现）
    async fn inner_run_command(
        &self,
        db_name: &str,
        command: Document,
    ) -> Result<Document> {
        let database = self.get_database(db_name);
        let result = database.run_command(command, None).await?;
        Ok(result)
    }
    
    /// 列出所有集合名称（内部实现）
    async fn inner_list_collections(&self, db_name: &str) -> Result<Vec<String>> {
        let database = self.get_database(db_name);
        let mut collections = Vec::new();
        
        let cursor = database.list_collection_names(None).await?;
        for result in cursor {
            collections.push(result);
        }
        
        Ok(collections)
    }
    
    /// 插入文档（内部实现）
    async fn inner_insert_one<T>(&self, db_name: &str, collection_name: &str, document: T) -> Result<()> 
    where
        T: serde::Serialize,
    {
        let collection = self.inner_get_collection(db_name, collection_name);
        collection.insert_one(document, None).await?;
        Ok(())
    }
    
    /// 批量插入文档（内部实现）
    async fn inner_insert_many<T>(&self, db_name: &str, collection_name: &str, documents: Vec<T>) -> Result<()> 
    where
        T: serde::Serialize,
    {
        let collection = self.inner_get_collection(db_name, collection_name);
        collection.insert_many(documents, None).await?;
        Ok(())
    }
    
    /// 查找文档（内部实现）
    async fn inner_find<T>(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Vec<T>> 
    where
        T: for<'de> serde::Deserialize<'de> + Unpin + Send + Sync,
    {
        let collection = self.inner_get_collection(db_name, collection_name);
        let cursor = collection.find(filter, None).await?;
        let results: Vec<T> = cursor.try_collect().await?;
        Ok(results)
    }
    
    /// 查找单个文档（内部实现）
    async fn inner_find_one<T>(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Option<T>> 
    where
        T: for<'de> serde::Deserialize<'de> + Unpin + Send + Sync,
    {
        let collection = self.inner_get_collection(db_name, collection_name);
        let result = collection.find_one(filter, None).await?;
        Ok(result)
    }
    
    /// 更新文档（内部实现）
    async fn inner_update_one(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
        update: Document,
    ) -> Result<()> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        collection.update_one(filter, update, None).await?;
        Ok(())
    }
    
    /// 删除文档（内部实现）
    async fn inner_delete_one(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
    ) -> Result<()> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        collection.delete_one(filter, None).await?;
        Ok(())
    }
    
    /// 聚合操作（内部实现，返回结果）
    async fn inner_aggregate_with_results(
        &self,
        db_name: &str,
        collection_name: &str,
        pipeline: Vec<Document>,
    ) -> Result<Vec<Document>> {
        let collection = self.inner_get_document_collection(db_name, collection_name);
        let cursor = collection.aggregate(pipeline, None).await?;
        let results: Vec<Document> = cursor.try_collect().await?;
        Ok(results)
    }
    
    /// 聚合操作（内部实现，不返回结果）
    async fn inner_aggregate(
        &self,
        db_name: &str,
        collection_name: &str,
        pipeline: Vec<Document>,
    ) -> Result<()> {
        let collection = self.inner_get_document_collection(db_name, collection_name);
        let mut cursor = collection.aggregate(pipeline, None).await?;
        
        // 消耗游标以确保聚合操作完成
        while cursor.advance().await? {}        
        Ok(())
    }
}

#[async_trait::async_trait]
impl IStorage for MongoStorage {
    fn get_document_collection(&self, db_name: &str, collection_name: &str) -> Box<dyn ICollection<Document>> {
        let db = self.get_database(db_name);
        let collection = db.collection::<Document>(collection_name);
        Box::new(MongoCollectionWrapper { collection })
    }
    
    async fn run_command(&self, db_name: &str, command: Document) -> Result<Document> {
        let db = self.get_database(db_name);
        let result = db.run_command(command, None).await?;
        Ok(result)
    }
    
    async fn list_collections(&self, db_name: &str) -> Result<Vec<String>> {
        let db = self.get_database(db_name);
        let mut collections = db.list_collections(None, None).await?;
        let mut collection_names = Vec::new();
        
        while let Some(collection) = collections.try_next().await? {
            collection_names.push(collection.name);
        }
        
        Ok(collection_names)
    }
    
    async fn insert_one_document(&self, db_name: &str, collection_name: &str, document: Document) -> Result<()> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        collection.insert_one(document, None).await?;
        Ok(())
    }
    
    async fn insert_many_documents(&self, db_name: &str, collection_name: &str, documents: Vec<Document>) -> Result<()> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        collection.insert_many(documents, None).await?;
        Ok(())
    }
    
    async fn find_documents(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Vec<Document>> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        let cursor = collection.find(filter, None).await?;
        let results: Vec<Document> = cursor.try_collect().await?;
        Ok(results)
    }
    
    async fn find_one_document(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Option<Document>> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        let result = collection.find_one(filter, None).await?;
        Ok(result)
    }
    
    async fn update_one(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
        update: Document,
    ) -> Result<()> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        collection.update_one(filter, update, None).await?;
        Ok(())
    }
    
    async fn update_many(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
        update: Document,
    ) -> Result<()> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        collection.update_many(filter, update, None).await?;
        Ok(())
    }
    
    async fn delete_one(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
    ) -> Result<()> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        collection.delete_one(filter, None).await?;
        Ok(())
    }
    
    async fn aggregate(
        &self,
        db_name: &str,
        collection_name: &str,
        pipeline: Vec<Document>,
    ) -> Result<Vec<Document>> {
        let collection = self.inner_get_collection::<Document>(db_name, collection_name);
        let cursor = collection.aggregate(pipeline, None).await?;
        let results: Vec<Document> = cursor.try_collect().await?;
        Ok(results)
    }
}

#[async_trait::async_trait]
impl IStorageExt for MongoStorage {
    fn get_collection<T>(&self, db_name: &str, collection_name: &str) -> Box<dyn ICollection<T>> where
        T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync + 'static {
        let db = self.get_database(db_name);
        let collection = db.collection::<T>(collection_name);
        Box::new(MongoCollectionWrapper { collection })
    }
    
    async fn insert_one<T>(&self, db_name: &str, collection_name: &str, document: T) -> Result<()>
    where
        T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync
    {
        let collection = self.inner_get_collection::<T>(db_name, collection_name);
        collection.insert_one(document, None).await?;
        Ok(())
    }
    
    async fn insert_many<T>(&self, db_name: &str, collection_name: &str, documents: Vec<T>) -> Result<()>
    where
        T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync
    {
        let collection = self.inner_get_collection::<T>(db_name, collection_name);
        collection.insert_many(documents, None).await?;
        Ok(())
    }
    
    async fn find<T>(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Vec<T>>
    where
        T: serde::Serialize + for<'de> serde::Deserialize<'de> + Unpin + Send + Sync + 'static
    {
        let collection = self.inner_get_collection::<T>(db_name, collection_name);
        let cursor = collection.find(filter, None).await?;
        let results: Vec<T> = cursor.try_collect().await?;
        Ok(results)
    }
    
    async fn find_one<T>(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Option<T>>
    where
        T: serde::Serialize + for<'de> serde::Deserialize<'de> + Unpin + Send + Sync + 'static
    {
        let collection = self.inner_get_collection::<T>(db_name, collection_name);
        let result = collection.find_one(filter, None).await?;
        Ok(result)
    }
}



#[cfg(test)]
mod tests {
    use super::*;
    use mongodb::bson::doc;

    #[tokio::test]
    async fn test_mongo_storage_creation() {
        // 这个测试需要实际的MongoDB连接
        // 在实际环境中测试
        assert!(true);
    }
}